package com.zfei.firstflink;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.*;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import scala.util.parsing.json.JSONObject;

import java.util.ArrayList;


/***
 *$com.zfei.TestWindow
 *@author:jiwenfei
 *
 *@version:
 *
 *createTime:2018/8/16:下午2:41
 *
 *description:
 */
public class TestWindow {
    public static void main(String[] args) throws  Exception{
        StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<String> ee= env.socketTextStream("localhost",9999);
        ee.filter(i->!i.trim().equals(""))
                .flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {
                    @Override
                    public void flatMap(String s, Collector<Tuple2<String,Integer>> collector) throws Exception {
                        for (String word :s.split(" ")){
                            if (!s.trim().equals("")){
                                collector.collect(new Tuple2<>(word,1));
                            }
                        }
                    }
                })
                .keyBy(ii->ii.f0)
                //.timeWindow(Time.seconds(20),Time.seconds(10))
                //.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
               // .window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(5)))
               .window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)))
                .trigger(new Trigger<Tuple2<String, Integer>, TimeWindow>() {
                    @Override
                    public TriggerResult onElement(Tuple2<String, Integer> stringIntegerTuple2, long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
                        System.out.println(l);
                         return TriggerResult.CONTINUE;
                    }

                    @Override
                    public TriggerResult onProcessingTime(long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
                        System.out.println(l);
                        return TriggerResult.CONTINUE;
                    }

                    @Override
                    public TriggerResult onEventTime(long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
                        System.out.println(l);
                        return TriggerResult.CONTINUE;
                    }

                    @Override
                    public void clear(TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
                        System.out.println(triggerContext.toString());

                    }
                })
               // .timeWindow(Time.seconds(5))
                .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> stringIntegerTuple2, Tuple2<String, Integer> t1) throws Exception {
                        return new Tuple2<>(stringIntegerTuple2.f0,stringIntegerTuple2.f1+1);
                    }
                })
                .print();
        env.execute("test window");
    }
//    public  Boolean isHasTheKeys(ArrayList<String >keys, JSONObject data){
//        return true;
//    }
//
//    public Boolean storeDataToMongDb(String ruleId,JSONObject msg){
//
//        if (isHasTheKeys(ruleId,....)){
//            //update
//            //changeMap(msg.data)
//        }
//    }

}
